/**
 * 
 */
package cn.edu.bjtu.alex.rewrite.components;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileFilter;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;


import cn.edu.bjtu.alex.rewrite.Context;
import cn.edu.bjtu.alex.rewrite.interfaces.DocumentAnalyzer;
import cn.edu.bjtu.alex.rewrite.interfaces.Term;
import cn.edu.bjtu.alex.rewrite.tools.LuceneDocumentAnalyzer;
import cn.edu.bjtu.alex.rewrite.tools.NamedThreadFactory;
import cn.edu.bjtu.alex.rewrite.tools.RegExNumberFilter;
import cn.edu.bjtu.alex.rewrite.tools.SingleWordTermFilter;
import cn.edu.bjtu.alex.rewrite.tools.TermFilter;
import cn.edu.bjtu.entity.IDocument;
import util.docparser.impl.HiveWechatParser;
import util.docparser.interfaces.Parser;

/**
 * 在这个情况下,我们的数据集存放层次:
 *
 *userData/
 *	c1/
 *	   1.txt
 * 	c2/
 * 	   1.txt
 * 	c3/
 * 	   1.txt
 *  .../
 *  <strong>注意:每个文件中,第一行是一个文档,与原来的不同,原来每个文件是一个文档,所以这里面只有一个文件</strong>
 * 内部类,EachLabelWordAnalysisWorker就是用来对每个文件夹下的每个文件进行分词并统计总文档数<br>
 * 这里注意下:线程池,CountDownLatch,以及AtomicLong这三个类的应用.原子类通过CAS操作降低线程同步代价<br>
 * @author alex
 *
 */
public class CollectWord extends BaseComponent{
	private ExecutorService executorService;
	CountDownLatch latch;
	AtomicLong al = new AtomicLong(0);
	private final Set<TermFilter> filters = new HashSet<TermFilter>();
	public CollectWord(Context ctx) {
		super(ctx);
	}

	public void fitInternal() {
		filters.add(new SingleWordTermFilter());
		filters.add(new RegExNumberFilter());
		File dir = ctx.getInputDir();
		//类文件夹 名字如果没改的话一般是c1 c2 c3等
		File []dirs = dir.listFiles(new FileFilter() {
			public boolean accept(File pathname) {
				return pathname.isDirectory();
			}
		});
		//这里判断 c1 c2文件夹中是不是空，是不是有文件
		int numClsses = dirs.length;
		latch = new CountDownLatch(numClsses); 
		executorService = Executors.newCachedThreadPool(new NamedThreadFactory("POOL"));
		for(File f : dirs){
			String label = f.getName();
			ctx.addLabel(label);
			LOG.info("Add label: label=" + label);
			executorService.submit(new EachLabelWordAnalysisWorker(f.getName()));
		}
		try {
			latch.await();
			executorService.shutdown();
			ctx.setTotalDocCount((int)al.get());	
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		stat();
	}
	//统计量
	private void stat() {
		LOG.info("STAT: totalDocCount=" + ctx.totalDocCount());
		LOG.info("STAT: labelCount=" + ctx.labelCount());
		//Map<label, Map<doc ,Map<word, term>>>
		Iterator<Entry<String, Map<String, Map<String, Term>>>> iter = ctx.termTableIterator();
		while(iter.hasNext()) {
			Entry<String, Map<String, Map<String, Term>>> entry = iter.next();
			if(entry.getValue()!=null){
				Iterator<Entry<String, Map<String, Term>>> docIter = entry.getValue().entrySet().iterator();
				int termCount = 0;
				while(docIter.hasNext()) {
					termCount += docIter.next().getValue().size();
				}
				LOG.info("STAT: label=" + entry.getKey() + ", docCount=" + entry.getValue().size() + ", termCount=" + termCount);
			}
		}
	}
	
	//一个线程去搜集一个文件下的文档
	private final class EachLabelWordAnalysisWorker extends Thread {
		
		private final String label;
		private final DocumentAnalyzer analyzer;
		private final Context ctx;
		public EachLabelWordAnalysisWorker(String label) {
			this.label = label;
			this.ctx = CollectWord.this.ctx;
			//分词器
			analyzer = new LuceneDocumentAnalyzer(ctx); 
		}
		
		@Override
		public void run() {
			try {
				File labelDir = new File(CollectWord.this.ctx.getInputDir(), label);
				File[] files = labelDir.listFiles(new FileFilter() {
					public boolean accept(File pathname) {
						return pathname.getAbsolutePath().endsWith(ctx.getFileExtensionName());
					}
				});
				if(files==null || files.length==0){
					//这个文件夹cx里面没有文件.
					ctx.removeLabel(label);
					return ;
				}
				int n = 0;
				for(File file : files) {
					analyze(label, file);
					++n;
				}
			} finally {
				latch.countDown();
			}
		}
		long lineNum = 0;
		protected void analyze(String label, File file) {
			BufferedReader br = null;
			Parser p = new HiveWechatParser();
			try{
				br = new BufferedReader(new FileReader(file));
				String line = "";
				while((line = br.readLine())!=null){
					try{
						IDocument d = p.parse(line);
						String docid = d.getID();
						Map<String, Term> terms = analyzer.analyze(d.getContent());
						// filter terms
						filterTerms(terms);
						
						// construct memory structure
						ctx.addTerms(label, docid, terms);
						// add inverted table as needed
						ctx.addTermsToInvertedTable(label, docid, terms);
						lineNum ++ ;
					}catch(Exception e){
						LOG.warn("===========Error format at "+lineNum +" in "+label+"==============");
					}
					
				}
				al.addAndGet(lineNum);
			}catch(Exception e){
				e.printStackTrace();
			}finally{
				try {br.close();} catch (IOException e) {}
			}
			
		}
	}
	protected void filterTerms(Map<String, Term> terms) {
		for(TermFilter filter : filters) {
			filter.filter(terms);
		}
	}

	/* (non-Javadoc)
	 * @see cn.edu.bjtu.alex.rewrite.components.BaseComponent#transformInternal()
	 */
	@Override
	protected void transformInternal() {
		// TODO Auto-generated method stub
		
	}

}
